package org.example.rabbitmq.workqueues;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

public class Worker {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.0.228");
        factory.setUsername("mq");
        factory.setPassword("123456");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        /**
         * 设置服务质量（QoS），确保通道每次只接收一个未确认的消息
         * 这个设置对于流量控制非常重要，可以防止消费者被大量消息“淹没”
         * 通过一次只处理一条消息的方式，确保每条消息在被正确处理和确认之后，才接收下一条消息
         */
        channel.basicQos(1); // 每次只接收一个未确认的消息（详细说明见下方）


        /**
         * (consumerTag, delivery) -> { ... } 是 Java 8 引入的 Lambda 表达式语法，
         * 用于简化接口实现，尤其是函数式接口（只有一个抽象方法的接口）。
         * 这段代码实现了 DeliverCallback 接口的 handleDelivery 方法。
         * 该接口是一个函数式接口，因此可以用 Lambda 表达式来代替传统的匿名内部类写法。
         * (consumerTag, delivery)：参数列表，对应 handleDelivery 方法的两个参数。
         * ->：表示 Lambda 表达式的开始，将参数与函数体分离。
         * { ... }：方法体，即 handleDelivery 的具体实现。
         *
         * 等价的传统写法:
         * DeliverCallback deliverCallback = new DeliverCallback() {
         *     @Override
         *     public void handleDelivery(String consumerTag, Delivery delivery) throws IOException {
         *         // 处理消息的逻辑
         *     }
         * };
         */
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");

            try {
                doWork(message);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.println(" [x] Done");
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }

        };
        boolean autoAck = false; // acknowledgment is covered below
        channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {
        });
    }

    private static void doWork(String task) throws InterruptedException {
        for (char ch : task.toCharArray()) {
            if (ch == '.') Thread.sleep(5000);
        }
    }
}
